package com.ruoyi.socket.client;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.socket.manager.HuobiConnectionManager;
import com.ruoyi.socket.manager.WebSocketUserManager;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.handshake.ServerHandshake;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.zip.GZIPInputStream;

/**
 * 火币WebSocket客户端
 * 处理WebSocket连接、消息接收和数据分发
 */
@Slf4j
public class HuobiWebSocketClient extends WebSocketClient {
    
    private static final long RECONNECT_INTERVAL = 5000; // 重连间隔，单位：毫秒
    private static final long HEARTBEAT_INTERVAL = 20000; // 心跳间隔，单位：毫秒
    private static final int CONNECTION_TIMEOUT = 10000; // 连接超时，单位：毫秒
    
    private Timer reconnectTimer;
    private Timer heartbeatTimer;
    private Timer connectionTimeoutTimer;
    private WebSocketUserManager webSocketUserManager;
    private HuobiConnectionManager huobiConnectionManager;
    private List<String> subscriptions;
    private volatile boolean isConnecting = false;
    
    public HuobiWebSocketClient(URI serverUri, Draft draft, List<String> subscriptions, 
                               WebSocketUserManager webSocketUserManager, 
                               HuobiConnectionManager huobiConnectionManager) {
        super(serverUri, draft);
        this.subscriptions = subscriptions;
        this.webSocketUserManager = webSocketUserManager;
        this.huobiConnectionManager = huobiConnectionManager;
        this.reconnectTimer = new Timer();
        this.heartbeatTimer = new Timer();
        this.connectionTimeoutTimer = new Timer();
        
        // 设置连接超时
        this.setConnectionLostTimeout(CONNECTION_TIMEOUT / 1000);
        
        // 检查是否需要设置代理
        setupProxy();
    }
    
    /**
     * 设置代理配置
     */
    private void setupProxy() {
        try {
            // 检查系统代理设置
            String httpProxy = System.getProperty("http.proxyHost");
            String httpProxyPort = System.getProperty("http.proxyPort");
            String httpsProxy = System.getProperty("https.proxyHost");
            String httpsProxyPort = System.getProperty("https.proxyPort");
            
            if (httpsProxy != null && httpsProxyPort != null) {
                // 使用HTTPS代理
                java.net.Proxy proxy = new java.net.Proxy(
                    java.net.Proxy.Type.HTTP, 
                    new java.net.InetSocketAddress(httpsProxy, Integer.parseInt(httpsProxyPort))
                );
                this.setProxy(proxy);
                log.info("火币WebSocket使用HTTPS代理: {}:{}", httpsProxy, httpsProxyPort);
            } else if (httpProxy != null && httpProxyPort != null) {
                // 使用HTTP代理
                java.net.Proxy proxy = new java.net.Proxy(
                    java.net.Proxy.Type.HTTP, 
                    new java.net.InetSocketAddress(httpProxy, Integer.parseInt(httpProxyPort))
                );
                this.setProxy(proxy);
                log.info("火币WebSocket使用HTTP代理: {}:{}", httpProxy, httpProxyPort);
            } else {
                log.info("火币WebSocket未配置代理，使用直连");
            }
        } catch (Exception e) {
            log.warn("设置火币WebSocket代理失败: {}", e.getMessage());
        }
    }
    
    @Override
    public boolean connectBlocking() throws InterruptedException {
        isConnecting = true;
        
        // 设置连接超时定时器
        connectionTimeoutTimer.schedule(new TimerTask() {
            @Override
            public void run() {
                if (isConnecting && !isOpen()) {
                    log.warn("火币WebSocket连接超时，尝试切换地址");
                    isConnecting = false;
                    close();
                    handleConnectionTimeout();
                }
            }
        }, CONNECTION_TIMEOUT);
        
        try {
            return super.connectBlocking(CONNECTION_TIMEOUT / 1000, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            isConnecting = false;
            handleConnectionTimeout();
            throw e;
        }
    }
    
    @Override
    public void connect() {
        isConnecting = true;
        
        // 设置连接超时定时器
        connectionTimeoutTimer.schedule(new TimerTask() {
            @Override
            public void run() {
                if (isConnecting && !isOpen()) {
                    log.warn("火币WebSocket连接超时，尝试切换地址");
                    isConnecting = false;
                    close();
                    handleConnectionTimeout();
                }
            }
        }, CONNECTION_TIMEOUT);
        
        super.connect();
    }
    
    private void handleConnectionTimeout() {
        if (huobiConnectionManager != null) {
            boolean shouldSwitch = huobiConnectionManager.handleConnectionFailure("连接超时");
            if (shouldSwitch) {
                log.info("火币WebSocket连接超时，切换到下一个地址");
                reconnectWithNewUrl();
            } else {
                log.warn("火币WebSocket连接超时，所有地址都已尝试，等待重连");
                reconnectWebSocket();
            }
        } else {
            reconnectWebSocket();
        }
    }
    
    @Override
    public void onOpen(ServerHandshake serverHandshake) {
        isConnecting = false;
        log.info("火币WebSocket连接已打开");
        
        if (huobiConnectionManager != null) {
            huobiConnectionManager.handleConnectionSuccess();
        }
        
        // 启动心跳
        startHeartbeat();
        
        // 订阅数据
        subscribeToChannels();
    }
    
    @Override
    public void onMessage(String message) {
        try {
            log.debug("收到火币WebSocket文本消息: {}", message);
            JSONObject jsonObject = JSON.parseObject(message);
            
            // 处理ping消息
            if (jsonObject.containsKey("ping")) {
                handlePing(jsonObject.getLong("ping"));
                return;
            }
            
            // 处理订阅确认
            if (jsonObject.containsKey("subbed")) {
                log.info("火币订阅成功: {}", jsonObject.getString("subbed"));
                return;
            }
            
            // 处理订阅错误
            if (jsonObject.containsKey("status") && "error".equals(jsonObject.getString("status"))) {
                log.error("火币订阅失败: {}", message);
                return;
            }
            
            // 处理数据推送
            if (jsonObject.containsKey("ch")) {
                handleMarketData(jsonObject);
            } else {
                log.debug("收到未识别的火币消息: {}", message);
            }
            
        } catch (Exception e) {
            log.error("处理火币WebSocket消息时出错: {}", message, e);
        }
    }
    
    @Override
    public void onMessage(ByteBuffer bytes) {
        try {
            log.debug("收到火币WebSocket二进制消息，大小: {} 字节", bytes.remaining());
            
            // 火币WebSocket数据是GZIP压缩的
            String decompressed = decompress(bytes.array());
            log.debug("解压缩后的消息: {}", decompressed);
            
            onMessage(decompressed);
        } catch (Exception e) {
            log.error("解压火币WebSocket消息时出错，消息大小: {} 字节", bytes.remaining(), e);
        }
    }
    
    @Override
    public void onClose(int code, String reason, boolean remote) {
        isConnecting = false;
        String errorMessage = "火币WebSocket连接关闭. Code: " + code + ", Reason: " + reason;
        log.error(errorMessage);
        
        // 停止心跳
        stopHeartbeat();
        
        if (huobiConnectionManager != null) {
            boolean shouldSwitch = huobiConnectionManager.handleConnectionFailure(errorMessage);
            if (shouldSwitch) {
                // 切换地址后重新连接
                reconnectWithNewUrl();
                return;
            }
        }
        
        if (code != -1) {
            reconnectWebSocket();
        }
    }
    
    @Override
    public void onError(Exception e) {
        isConnecting = false;
        String errorMessage = "火币WebSocket错误: " + e.getMessage();
        log.error(errorMessage, e);
        
        // 停止心跳
        stopHeartbeat();
        
        if (huobiConnectionManager != null) {
            boolean shouldSwitch = huobiConnectionManager.handleConnectionFailure(errorMessage);
            if (shouldSwitch) {
                // 切换地址后重新连接
                reconnectWithNewUrl();
                return;
            }
        }
        
        reconnectWebSocket();
    }
    
    /**
     * 处理ping消息
     */
    private void handlePing(long pingTime) {
        JSONObject pong = new JSONObject();
        pong.put("pong", pingTime);
        send(pong.toString());
        log.debug("发送pong响应: {}", pingTime);
    }
    
    /**
     * 处理市场数据
     */
    private void handleMarketData(JSONObject jsonObject) {
        String channel = jsonObject.getString("ch");
        
        if (channel.contains("kline")) {
            // K线数据
            handleKlineData(jsonObject);
        } else if (channel.contains("trade")) {
            // 交易数据
            handleTradeData(jsonObject);
        } else if (channel.contains("detail")) {
            // 详情数据 - 24小时统计
            handleDetailData(jsonObject);
        } else if (channel.contains("ticker")) {
            // 聚合行情数据
            handleTickerData(jsonObject);
        } else {
            log.debug("收到未处理的火币数据类型: {}", channel);
        }
    }
    
    /**
     * 处理K线数据
     */
    private void handleKlineData(JSONObject jsonObject) {
        try {
            String channel = jsonObject.getString("ch");
            String symbol = parseSymbolFromChannel(channel);
            log.debug("收到火币K线数据: {} -> {}", symbol, jsonObject.toString().substring(0, Math.min(200, jsonObject.toString().length())));
            
            if (webSocketUserManager != null) {
                webSocketUserManager.huobiKlineSendMsg(jsonObject.toString());
            }
        } catch (Exception e) {
            log.error("处理火币K线数据时出错", e);
        }
    }
    
    /**
     * 处理交易数据
     */
    private void handleTradeData(JSONObject jsonObject) {
        try {
            String channel = jsonObject.getString("ch");
            String symbol = parseSymbolFromChannel(channel);
            log.debug("收到火币交易数据: {} -> {}", symbol, jsonObject.toString().substring(0, Math.min(200, jsonObject.toString().length())));
            
            if (webSocketUserManager != null) {
                webSocketUserManager.huobiTradeSendMsg(jsonObject.toString());
            }
        } catch (Exception e) {
            log.error("处理火币交易数据时出错", e);
        }
    }
    
    /**
     * 处理详情数据 - 24小时统计数据
     */
    private void handleDetailData(JSONObject jsonObject) {
        try {
            String channel = jsonObject.getString("ch");
            String symbol = parseSymbolFromChannel(channel);
            
            // 提取关键价格信息用于日志
            JSONObject tick = jsonObject.getJSONObject("tick");
            String priceInfo = "";
            if (tick != null) {
                priceInfo = String.format("价格:%s, 变化:%.2f%%", 
                    tick.getString("close"), 
                    calculatePriceChangePercent(tick.getString("open"), tick.getString("close")));
            }
            
            log.debug("收到火币详情数据: {} -> {}", symbol, priceInfo);
            
            if (webSocketUserManager != null) {
                // 确保火币详情数据被正确处理和分发
                webSocketUserManager.huobiDetailSendMsg(jsonObject.toString());
            }
        } catch (Exception e) {
            log.error("处理火币详情数据时出错", e);
        }
    }
    
    /**
     * 处理聚合行情数据
     */
    private void handleTickerData(JSONObject jsonObject) {
        try {
            String channel = jsonObject.getString("ch");
            String symbol = parseSymbolFromChannel(channel);
            log.debug("收到火币聚合行情数据: {} -> {}", symbol, jsonObject.toString().substring(0, Math.min(200, jsonObject.toString().length())));
            
            // 聚合行情数据也可以用于详情显示
            if (webSocketUserManager != null) {
                webSocketUserManager.huobiDetailSendMsg(jsonObject.toString());
            }
        } catch (Exception e) {
            log.error("处理火币聚合行情数据时出错", e);
        }
    }
    
    /**
     * 计算价格变化百分比
     */
    private double calculatePriceChangePercent(String openPrice, String closePrice) {
        try {
            BigDecimal open = new BigDecimal(openPrice);
            BigDecimal close = new BigDecimal(closePrice);
            if (open.compareTo(BigDecimal.ZERO) > 0) {
                return close.subtract(open).divide(open, 4, RoundingMode.HALF_UP)
                          .multiply(new BigDecimal("100")).doubleValue();
            }
        } catch (Exception e) {
            // 忽略计算错误
        }
        return 0.0;
    }
    
    /**
     * 从频道名称解析币种
     */
    private String parseSymbolFromChannel(String channel) {
        if (channel != null && channel.startsWith("market.")) {
            String[] parts = channel.split("\\.");
            if (parts.length >= 2) {
                String pair = parts[1]; // btcusdt
                if (pair.endsWith("usdt")) {
                    return pair.substring(0, pair.length() - 4).toLowerCase(); // btc
                }
            }
        }
        return "unknown";
    }
    
    /**
     * 订阅频道 - 优化订阅逻辑
     */
    private void subscribeToChannels() {
        CompletableFuture.runAsync(() -> {
            try {
                log.info("开始订阅火币频道，总数: {}", subscriptions.size());
                Thread.sleep(2000); // 等待连接完全稳定
                
                int successCount = 0;
                int failureCount = 0;
                
                for (String subscription : subscriptions) {
                    try {
                        JSONObject subMsg = new JSONObject();
                        subMsg.put("sub", subscription);
                        subMsg.put("id", System.currentTimeMillis() + "_" + subscription.hashCode());
                        
                        String subscriptionMessage = subMsg.toString();
                        log.info("发送火币订阅消息: {}", subscriptionMessage);
                        
                        send(subscriptionMessage);
                        successCount++;
                        
                        // 避免发送过快，每个订阅间隔200ms
                        Thread.sleep(200);
                        
                    } catch (Exception e) {
                        failureCount++;
                        log.error("订阅火币频道失败: {}", subscription, e);
                    }
                }
                
                log.info("火币频道订阅完成 - 成功: {}, 失败: {}", successCount, failureCount);
                
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.error("订阅频道时被中断", e);
            } catch (Exception e) {
                log.error("订阅频道时发生异常", e);
            }
        });
    }
    
    /**
     * 启动心跳
     */
    private void startHeartbeat() {
        heartbeatTimer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                if (isOpen()) {
                    // 火币不需要主动发送心跳，只需要响应ping
                    log.debug("火币WebSocket连接正常");
                }
            }
        }, HEARTBEAT_INTERVAL, HEARTBEAT_INTERVAL);
    }
    
    /**
     * 停止心跳
     */
    private void stopHeartbeat() {
        if (heartbeatTimer != null) {
            heartbeatTimer.cancel();
            heartbeatTimer = new Timer();
        }
    }
    
    /**
     * 重连WebSocket
     */
    private void reconnectWebSocket() {
        reconnectTimer.schedule(new TimerTask() {
            @Override
            public void run() {
                try {
                    log.info("尝试重新连接火币WebSocket...");
                    reconnect();
                } catch (Exception e) {
                    log.error("重连火币WebSocket失败", e);
                }
            }
        }, RECONNECT_INTERVAL);
    }
    
    /**
     * 使用新URL重连
     */
    private void reconnectWithNewUrl() {
        if (huobiConnectionManager != null) {
            reconnectTimer.schedule(new TimerTask() {
                @Override
                public void run() {
                    try {
                        String newUrl = huobiConnectionManager.getCurrentWebSocketUrl();
                        log.info("使用新地址重新连接火币WebSocket: {}", newUrl);
                        
                        // 关闭当前连接
                        close();
                        
                        // 重新连接
                        reconnect();
                    } catch (Exception e) {
                        log.error("使用新地址重连失败", e);
                        reconnectWebSocket();
                    }
                }
            }, RECONNECT_INTERVAL);
        } else {
            reconnectWebSocket();
        }
    }
    
    /**
     * 解压GZIP数据
     */
    private String decompress(byte[] data) throws Exception {
        try (ByteArrayInputStream bis = new ByteArrayInputStream(data);
             GZIPInputStream gzis = new GZIPInputStream(bis);
             ByteArrayOutputStream bos = new ByteArrayOutputStream()) {
            
            byte[] buffer = new byte[1024];
            int len;
            while ((len = gzis.read(buffer)) != -1) {
                bos.write(buffer, 0, len);
            }
            
            return bos.toString("UTF-8");
        }
    }
    
    /**
     * 关闭连接和定时器
     */
    @Override
    public void close() {
        stopHeartbeat();
        if (reconnectTimer != null) {
            reconnectTimer.cancel();
        }
        super.close();
    }
} 